package com.hskn.hss.config;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hskn.hss.module.kfkanalyse.service.IkfkAnalyseService;
import com.hskn.hss.module.kfkcarvideo.mapper.KfkCarVideoMapper;
import com.hskn.hss.module.kfkdeal.service.IkfkDealService;
import com.hskn.hss.module.kfkdeal.to.KfkDealEmployeeTo;
import com.hskn.hss.module.kfkdtu.mapper.KfkDtuMapper;
import com.hskn.hss.module.kfkhumancamer.mapper.KfkHumanCamerMapper;
import com.hskn.hss.module.kfkinterphone.entity.KfkInterphone;
import com.hskn.hss.module.kfkinterphone.mapper.KfkInterphoneMapper;
import com.hskn.hss.module.kfknvr.mapper.KfkNvrMapper;
import com.hskn.hss.module.kfkpunchcard.mapper.KfkPunchCardMapper;
import com.hskn.hss.module.kfkworkcard.mapper.KfkWorkCardMapper;
import com.hskn.hss.module.tlemployee.entity.TlEmployee;
import com.hskn.hss.module.tlemployee.mapper.TlEmployeeMapper;
import com.hskn.hss.module.tlequip.entity.TlEquip;
import com.hskn.hss.module.tlequip.mapper.TlEquipMapper;
import com.hskn.hss.utils.KafkaUtils;
import com.hskn.hss.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Date;
import java.util.Map;

/**
 * @author lny
 * createTime 2021-09-09
 **/
@Order(value = 5)
@Slf4j
@Component
public class KfkInterphoneRunner implements CommandLineRunner {

    @Autowired
    private IkfkDealService ikfkDealService;
    @Autowired
    private TlEquipMapper tlEquipMapper;
    @Autowired
    TlEmployeeMapper tlEmployeeMapper;
    @Autowired
    KfkWorkCardMapper kfkWorkCardMapper;
    @Autowired
    IkfkAnalyseService ikfkAnalyseService;
    @Autowired
    KfkInterphoneMapper kfkInterphoneMapper;
    @Autowired
    KfkPunchCardMapper kfkPunchCardMapper;
    @Autowired
    KfkHumanCamerMapper kfkHumanCamerMapper;
    @Autowired
    KfkDtuMapper kfkDtuMapper;
    @Autowired
    KfkCarVideoMapper kfkCarVideoMapper;
    @Autowired
    KfkNvrMapper kfkNvrMapper;

    @Value("${kafka.interphone_topic}")
    private String interphone_topic;

    @Value("${kafka.interphone_group}")
    private String interphone_group;

    @Value("${kafka.kafka_receive:true}")
    private Boolean kafka_receive;

    @Override
    public void run(String... args) {
        if (!kafka_receive) {
            return;
        }
        new Thread(() -> {
            KafkaConsumer consumer = KafkaUtils.getConsumer(interphone_group);
            consumer.subscribe(Arrays.asList(interphone_topic));
            while (true) {
                receiveAndHandle(consumer);
            }
        }).start();
    }

    private void receiveAndHandle(KafkaConsumer consumer) {
        //拉取kafka数据
        log.info("====================kfk定时读取任务开始-对讲机======================================");
        try {
            ConsumerRecords<Object, Object> records = consumer.poll(1000);
            log.info("========KfkInterphoneRunner拉取条数：{}", records.count());
            for (ConsumerRecord<Object, Object> record : records) {
                try {
                    Map value = JSONObject.parseObject(String.valueOf(record.value()), Map.class);
                    Map content = JSONObject.parseObject(String.valueOf(value.get("content")), Map.class);
                    Map info = JSONObject.parseObject(String.valueOf(content.get("info")), Map.class);
                    Date timestamp = new Date(record.timestamp());
                    log.info("KfkInterphoneRunner0001:if (value.get(\"deviceType\").equals(\"INTERPHONE\") && content.get(\"type\").equals(\"INTERPHONE_REPORT\")) {");
                    if (value.get("deviceType").equals("INTERPHONE") && content.get("type").equals("INTERPHONE_REPORT")) { //对讲机
                        long interphoneStartTime = System.currentTimeMillis();
                        Long interphoneTimeBefore = System.currentTimeMillis();
                        Date date = StringUtils.dealDateFormat(String.valueOf(info.get("Time")));
                        System.out.println(DateUtil.formatDateTime(date));
                        KfkInterphone kfkInterphone = new KfkInterphone();
                        kfkInterphone.setEqpnum(String.valueOf(value.get("no")));
                        kfkInterphone.setState(String.valueOf(value.get("state")));
                        kfkInterphone.setTimestamp(timestamp);
                        kfkInterphone.setDeviceTime(date);
                        kfkInterphone.setLat(String.valueOf(info.get("Lat")));
                        kfkInterphone.setLon(String.valueOf(info.get("Lon")));
                        kfkInterphone.setDirection(Double.parseDouble(info.get("Direction").toString()));
                        kfkInterphone.setSpeed(Double.parseDouble(info.get("Speed").toString()));
                        kfkInterphone.setAccount(String.valueOf(info.get("Account")));
                        kfkInterphone.setCreateTime(new Date());
                        kfkInterphoneMapper.insert(kfkInterphone);
                        log.info("KfkInterphoneRunner0002:kfkInterphoneMapper.insert(kfkInterphone);");
                        QueryWrapper<TlEquip> queryWrapper = new QueryWrapper<>();
                        queryWrapper.lambda().eq(TlEquip::getEquipcode, kfkInterphone.getEqpnum());
                        TlEquip tlEquip = tlEquipMapper.selectOne(queryWrapper);
                        log.info("KfkInterphoneRunner0003:TlEquip tlEquip = tlEquipMapper.selectOne(queryWrapper);");
                        QueryWrapper<TlEmployee> queryWrapper2 = new QueryWrapper<>();
                        queryWrapper2.lambda().eq(TlEmployee::getId, tlEquip.getBindobjid());
                        TlEmployee tlEmployee = tlEmployeeMapper.selectOne(queryWrapper2);
                        log.info("KfkInterphoneRunner0004:TlEmployee tlEmployee = tlEmployeeMapper.selectOne(queryWrapper2);");
                        if (null != tlEmployee) {
                            tlEmployee.setAddr(kfkInterphone.getLon() + "," + kfkInterphone.getLat());
                            tlEmployeeMapper.updateById(tlEmployee);
                        }
                        ikfkAnalyseService.KfkInterphoneStayAnalyse(kfkInterphone);
                        log.info("KfkInterphoneRunner0005:ikfkAnalyseService.KfkInterphoneStayAnalyse(kfkInterphone);");
                        KfkDealEmployeeTo kfkDealEmployeeTo = new KfkDealEmployeeTo();
                        kfkDealEmployeeTo.setEqp_type(2);
                        kfkDealEmployeeTo.setDate(date);
                        kfkDealEmployeeTo.setLng(kfkInterphone.getLon());
                        kfkDealEmployeeTo.setLat(kfkInterphone.getLat());
                        kfkDealEmployeeTo.setEquanum(kfkInterphone.getEqpnum());
                        kfkDealEmployeeTo.setEmployee_id(tlEquip.getBindobjid());
                        kfkDealEmployeeTo.setSpeed(kfkInterphone.getSpeed().intValue());
                        kfkDealEmployeeTo.setDirection(kfkInterphone.getDirection().intValue());
                        ikfkDealService.KfkDealEmployee(kfkDealEmployeeTo);
                        log.info("KfkInterphoneRunner0006: ikfkDealService.KfkDealEmployee(2, date, kfkInterphone.getLon(), kfkInterphone.getLat(), kfkInterphone.getAccount(), tlEquip.getBindobjid(), kfkInterphone.getSpeed().intValue(), kfkInterphone.getDirection().intValue());");
                        Long interphoneTimeAfter = System.currentTimeMillis();
                        Long cla = interphoneTimeAfter - interphoneTimeBefore;
                        log.info("===INTERPHONE,time:{}", cla);
                        long interphoneEndTime = System.currentTimeMillis();
                        log.info("=====对讲机=执行时间(毫秒)：{}", interphoneEndTime - interphoneStartTime);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
            log.info("====================kfk定时读取任务结束-对讲机======================================");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            consumer.commitAsync();
        }
    }
}